package com.study.utils;

import com.study.entity.OrderEvent;
import com.study.entity.PayEvent;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * @author sjw
 * @className KafkaUtils
 * @date 2020/11/20 8:55
 * @description: TODO
 */
public class KafkaCheckOrderPayUtils {

    public static final String broker_list = "192.168.10.200:9092";
    public static final String orderTopic = "orderEvent";  // kafka topic，Flink 程序中需要和这个统一
    public static final String payTopic = "payEvent";
    public static final String[] orderAction = new String[]{"create", "pay"};
    public static final String[] payChannel = new String[]{"wechat", "alipay"};


    public static KafkaProducer getKafkaProducer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", broker_list);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //key 序列化
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //value 序列化
        KafkaProducer producer = new KafkaProducer<String, String>(props);
        return producer;
    }

    public static void writeToKafka(KafkaProducer producer, String topic) throws InterruptedException {
        String msgStr = "";


        long ts = System.currentTimeMillis();
        if (orderTopic.equals(topic)) {
            OrderEvent orderEvent = new OrderEvent();
            orderEvent.setUserId(ts % 10000 + 1);
            orderEvent.setOrderId("order" + ts);
            orderEvent.setAction(orderAction[(int) (ts % 2)]);
            orderEvent.setTimestamp(ts);
            msgStr = JsonHelper.objectToJson(orderEvent);
        } else {
            PayEvent payEvent = new PayEvent();
            payEvent.setOrderId("order" + ts);
            payEvent.setPayChannel(payChannel[(int) (ts % 2)]);
            payEvent.setTimestamp(ts);
            msgStr = JsonHelper.objectToJson(payEvent);
        }
        ProducerRecord record = new ProducerRecord<String, String>(topic, null, null, msgStr);
        producer.send(record);
        System.out.println("发送数据: " + msgStr);

        producer.flush();
    }


    public static void main(String[] args) throws InterruptedException {
        KafkaProducer producer = getKafkaProducer();

        long ts = System.currentTimeMillis();
        for (int i = 0; i < 2; i++) {
            writeToKafka(producer, orderTopic);
            //writeToKafka(producer, payTopic);
        }

        writeToKafka(producer, orderTopic);
        Thread.sleep(5000);
        writeToKafka(producer, payTopic);
        //writeToKafka(producer);
    }
}
